package com.ds.lens.flink.monitor.datasource.function;

import com.alibaba.fastjson.JSONObject;
import com.ds.lens.data.common.document.datasource.LensDsSchema;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.types.Row;

import java.io.IOException;
import java.util.List;

/**
 * @author WeiShaoying
 * @date 2020/4/26 下午3:28
 */
@Slf4j
public class KafkaDataSource implements DeserializationSchema<Row> {

    private RowTypeInfo rowTypeInfo;
    private List<LensDsSchema> schemas;

    public KafkaDataSource(RowTypeInfo rowTypeInfo, List<LensDsSchema> schemas) {
        this.rowTypeInfo = rowTypeInfo;
        this.schemas = schemas;
    }

    @Override
    public TypeInformation<Row> getProducedType() {
        return rowTypeInfo;
    }

    @Override
    public Row deserialize(byte[] message) throws IOException {
        Row row = new Row(schemas.size());
        try {
            JSONObject value = JSONObject.parseObject(new String(message));
            for (int i = 0; i < schemas.size(); i++) {
                if (schemas.get(i).getSource().contains(".")) {
                    row.setField(i, getValue(value.getJSONObject(schemas.get(i).getSource().split("\\.")[0]),
                            schemas.get(i).getSource().split("\\.")[1], schemas.get(i).getType()));
                } else {
                    row.setField(i, getValue(value, schemas.get(i).getSource(), schemas.get(i).getType()));
                }
            }
        } catch (Exception e) {
            log.error("KafkaDataSource,exception{}", e);
        }
        return row;
    }

    private Object getValue(JSONObject json, String name, String type) {
        switch (type) {
            case "String":
                return json.getString(name);
            case "Long":
                return json.getLong(name);
            case "Timestamp":
                return json.getTimestamp(name);
            default:
                return null;
        }
    }

    @Override
    public boolean isEndOfStream(Row nextElement) {
        return false;
    }
}
